/*
 * FileName: LimitedLongDeserializer.java
 * Author:   zzw
 * Date:     2019年01月23日
 * Description:
 */
package com.chezhibao.flink.deserializer;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataInputViewStreamWrapper;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;

import java.io.ByteArrayInputStream;
import java.io.IOException;

/**
 * 〈〉<br>
 * 〈〉
 *
 * @author zzw
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本]（可选）
 */
public class LimitedLongDeserializer implements KeyedDeserializationSchema<String> {

    private static final long serialVersionUID = 6966177118923713521L;
    private final TypeInformation<String> ti;
    private final TypeSerializer<String> ser;
    long cnt = 0;

    public LimitedLongDeserializer() {
        this.ti = Types.STRING;
        this.ser = ti.createSerializer(new ExecutionConfig());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return ti;
    }

    @Override
    public String deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {
        cnt++;
        DataInputView in = new DataInputViewStreamWrapper(new ByteArrayInputStream(message));
        String e = ser.deserialize(in);
        return e;
    }

    @Override
    public boolean isEndOfStream(String nextElement) {
        return cnt > 1110L;
    }
}
